package org.study.r2dbc.boot.service;

import io.r2dbc.spi.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.study.r2dbc.boot.model.User;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.math.BigDecimal;
import java.util.function.BiFunction;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2025年04月02日
 */
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
@Service
public class UserServiceImplByConnectionFactory implements UserServiceByConnectionFactory {
    private final ConnectionFactory connectionFactory;

    /**
     * Row到User的映射函数
     */
    private final BiFunction<Row, RowMetadata, User> userMapper = (row, rowMetadata) -> {
        User user = new User();
        user.setId(row.get("id", Long.class));
        user.setName(row.get("name", String.class));
        user.setAge(row.get("age", Integer.class));
        user.setEmail(row.get("email", String.class));
        user.setBalance(new BigDecimal(row.get("balance", String.class)));
        user.setActive(row.get("active", Boolean.class));
        return user;
    };

    @Override
    public Mono<User> insertUser(User user) {
        return Mono.usingWhen(connectionFactory.create(), connection -> {
                    Statement statement = connection.createStatement("INSERT INTO t_user(id, name, age, email, balance, active) VALUES ($1, $2, $3, $4, $5, $6)");
                    statement.bind(0, user.getId())
                            .bind(1, user.getName())
                            .bind(2, user.getAge())
                            .bind(3, user.getEmail())
                            .bind(4, user.getBalance())
                            .bind(5, user.getActive());
                    return Mono.from(statement.execute()).flatMap(result -> Mono.from(result.map(userMapper)));
                },
                Connection::close
        );
    }

    @Override
    public Flux<User> findAll() {
        return Flux.usingWhen(connectionFactory.create(), connection ->
                        Mono.from(connection.createStatement("SELECT * FROM t_user").execute())
                                .flatMapMany(result -> Flux.from(result.map(userMapper))),
                Connection::close
        );
    }

    @Override
    public Mono<User> findById(Long id) {
        return Mono.usingWhen(connectionFactory.create(), connection -> {
                    Statement statement = connection.createStatement("SELECT * FROM t_user WHERE id = $1");
                    statement.bind(0, id);
                    return Mono.from(statement.execute()).flatMap(result -> Mono.from(result.map(userMapper)));
                },
                Connection::close
        );
    }

    @Override
    public Mono<User> update(User user) {
        return Mono.usingWhen(connectionFactory.create(), connection -> {
                    Statement statement = connection.createStatement("UPDATE t_user SET name = $1, age = $2, email = $3, balance = $4, active = $5 WHERE id = $6");
                    statement.bind(0, user.getName())
                            .bind(1, user.getAge())
                            .bind(2, user.getEmail())
                            .bind(3, user.getBalance())
                            .bind(4, user.getActive())
                            .bind(5, user.getId());
                    return Mono.from(statement.execute()).flatMap(result -> Mono.from(result.map(userMapper)));
                },
                Connection::close
        );
    }

    @Override
    public Mono<Boolean> delete(Long id) {
        return Mono.usingWhen(connectionFactory.create(), connection -> {
                    Statement statement = connection.createStatement("DELETE FROM t_user WHERE id = $1");
                    statement.bind(0, id);
                    return Mono.from(statement.execute()).flatMap(result -> Mono.from(result.getRowsUpdated()).map(rows -> rows > 0));
                },
                Connection::close
        );
    }
}
